package avicit.bdp.dlm.job;

import avicit.bdp.common.datasource.BaseDataSource;
import avicit.bdp.common.service.dto.DataSourceDTO;
import avicit.bdp.common.service.service.DataSourceService;
import avicit.bdp.common.utils.DateUtils;
import avicit.bdp.common.utils.EntityUtils;
import avicit.bdp.common.utils.database.DBUtils;
import avicit.bdp.common.utils.database.adapter.DbAdapter;
import avicit.bdp.common.utils.database.adapter.DbAdapterUtils;
import avicit.bdp.common.utils.database.adapter.impl.HiveDBAdapter;
import avicit.bdp.common.utils.redis.RedisLockUtils;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.dlm.dto.DlmDataSourceDTO;
import avicit.bdp.dlm.dto.DlmHiveStatDTO;
import avicit.bdp.dlm.dto.DlmStatDTO;
import avicit.bdp.dlm.service.DlmDataSourceService;
import avicit.bdp.dlm.service.DlmHiveStatService;
import avicit.bdp.dlm.service.DlmStatService;
import avicit.platform6.commons.utils.ComUtil;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.sql.Connection;
import java.util.List;

/**
 * @author xugb
 */
@Component
public class DlmJob {
    private static final Logger logger = LoggerFactory.getLogger(DlmJob.class);

    @Autowired
    private RedisLockUtils lockUtils;

    @Autowired
    private DataSourceService dataSourceService;

    @Autowired
    private DlmStatService dlmStatService;

    @Autowired
    private DlmDataSourceService dlmDataSourceService;

    @Autowired
    private DlmHiveStatService dlmHiveStatService;

    private static final String STAT_LOCK = "bdp:dlm:stat-dlm-data";

    @Scheduled(cron = "${scheduler.dmlStat:0 30 0 * * ?}")
    public void statData() throws Exception {

        boolean lock = lockUtils.lock(STAT_LOCK);
        if (lock) {
            logger.info("开始执行数据源统计。");
            try {
                long startTime = System.currentTimeMillis();
                //获取所有接入的数据源
                List<DataSourceDTO> dataSourceList = dataSourceService.queryDsByStat();
                if (CollectionUtils.isNotEmpty(dataSourceList)) {
                    //获取昨天日期
                    String yesterday = DateUtils.getDayNumBeforeToday(1);
                    for (DataSourceDTO dataSourceDTO : dataSourceList) {
                        DlmStatDTO dlmStatDTO = new DlmStatDTO();
                        dlmStatDTO.setStatDate(yesterday);
                        dlmStatDTO.setId(ComUtil.getId());
                        dlmStatDTO.setDataSourceId(dataSourceDTO.getId());
                        dlmStatDTO.setDataSourceStructure(dataSourceDTO.getDatasourceStructure());
                        try {

                            try {
                                handleUsable(dataSourceDTO, dlmStatDTO);
                            } catch (Exception e) {
                                logger.error(e.getMessage(), e);
                                //不可用数据暂时统计为0
                                handleDisabled(dlmStatDTO);
                            }
                            DlmStatDTO query = new DlmStatDTO();
                            query.setStatDate(yesterday);
                            query.setDataSourceId(dataSourceDTO.getId());
                            List<DlmStatDTO> list = dlmStatService.selectList(query);
                            //判断重复增加
                            if (list.isEmpty()) {
                                dlmStatService.insertSelective(dlmStatDTO);
                            } else {
                                dlmStatDTO.setId(list.get(0).getId());
                                dlmStatService.updateByPrimaryKeySelective(dlmStatDTO);
                            }

                        } catch (Exception e) {
                            logger.error(dataSourceDTO.getDatasourceName() + "统计出错：" + e.getMessage(), e);
                        }
                    }
                }

                long total = System.currentTimeMillis() - startTime;
                logger.info("结束统计工作台运行情况: 共用时：{}", total / 1000);
            } catch (Exception e) {
                logger.info("执行数据源统计执行错误。", e);
            } finally {
                lockUtils.delete(STAT_LOCK);
            }
        }

    }


    /**
     * 处理可用数据源连接
     *
     * @param dataSourceDTO 数据源
     * @param dlmStatDTO    统计数据
     * @author felix
     * @date 2020/11/30 15:27
     */
    private void handleUsable(DataSourceDTO dataSourceDTO, DlmStatDTO dlmStatDTO) throws Exception {
        //获取对应的适配器
        DbAdapter adapter = DbAdapterUtils.getDbAdapter(dataSourceDTO.getDatasourceType());
        //获取数据源版本，存到dlmdatasource表中
        BaseDataSource baseDataSource = DBUtils.getBaseDataSource(dataSourceDTO);
        DlmDataSourceDTO dlmQuery = new DlmDataSourceDTO();
        dlmQuery.setDataSourceId(dataSourceDTO.getId());
        DlmDataSourceDTO dlmDataSourceDTO = dlmDataSourceService.selectOne(dlmQuery);
        String versionInfo;
        try {
            versionInfo = adapter.getVersion(baseDataSource);
        } catch (Exception e) {
            versionInfo = "";
        }
        dlmDataSourceDTO.setDataSourceVersion(versionInfo);
        dlmDataSourceService.updateByPrimaryKeySelective(dlmDataSourceDTO);
        //获取数据源已用容量
        Long totalAmount = adapter.getDatabaseSize(dataSourceDTO);
        //每日增量
        //1得到最近一天的数据
        String dataSourceId = dataSourceDTO.getId();
        Long historyAmount = dlmStatService.selectLatestAmount(dataSourceId);
        if (historyAmount == null) {
            historyAmount = 0L;
        }
        Long dayAmount = totalAmount - historyAmount;
        //获取表数、文件数
        Long count = adapter.getTableCount(dataSourceDTO);
        //todo 存储组、桶
        Long otherCount = adapter.getOtherCount(dataSourceDTO);
        //获取数据源总容量
        Long availableAmount = 0L;
        DlmDataSourceDTO query = new DlmDataSourceDTO();
        query.setDataSourceId(dataSourceDTO.getId());
        List<DlmDataSourceDTO> dlmDataSourceDTOS = dlmDataSourceService.selectList(query);
        if (CollectionUtils.isNotEmpty(dlmDataSourceDTOS)) {
            availableAmount = dlmDataSourceDTOS.get(0).getAvailableAmount();
            if ((availableAmount == null || availableAmount == 0) &&
                    ("oracle".equals(dataSourceDTO.getDatasourceType())
                            || "hdfs".equals(dataSourceDTO.getDatasourceType()))) {
                availableAmount = adapter.getAvailableAmount(baseDataSource);
            }
        }
        dlmStatDTO.setTotalAmount(totalAmount);
        dlmStatDTO.setDayAmount(dayAmount);
        dlmStatDTO.setObjectCount(count);
        dlmStatDTO.setOtherCount(otherCount);
        dlmStatDTO.setAvailableAmount(availableAmount);
        EntityUtils.setCreateAndUpdateInfo(dlmStatDTO);
        /*******************处理hive数据源表数据量********************/
        if (Constants.DB_TYPE_HIVE.equals(dataSourceDTO.getDatasourceType())) {
            Connection conn = null;
            HiveDBAdapter hiveAdapter = (HiveDBAdapter) DbAdapterUtils.getDbAdapter(Constants.DB_TYPE_HIVE);
            try {
                conn = DBUtils.buildConnection(dataSourceDTO);
                List<String> tableNames = hiveAdapter.showTables(conn);
                if (CollectionUtils.isNotEmpty(tableNames)) {
                    for (String tableName : tableNames) {
                        DlmHiveStatDTO dlmHiveStatDTO = new DlmHiveStatDTO();
                        dlmHiveStatDTO.setId(ComUtil.getId());
                        dlmHiveStatDTO.setStatDate(dlmStatDTO.getStatDate());
                        dlmHiveStatDTO.setDataSourceId(dlmStatDTO.getDataSourceId());
                        dlmHiveStatDTO.setTableName(tableName);
                        Long tableAmount = hiveAdapter.getTotalSize(tableName, conn);
                        dlmHiveStatDTO.setTableAmount(tableAmount);
                        EntityUtils.setCreateAndUpdateInfo(dlmHiveStatDTO);
                        dlmHiveStatService.insert(dlmHiveStatDTO);
                    }
                }
            }finally {
                DBUtils.closeResource(conn,null,null);
            }
        }
    }

    /**
     * 处理不可用数据源连接，数据统计为0
     *
     * @param dlmStatDTO dlmStatDTO
     * @author felix
     * @date 2020/11/30 15:50
     */
    private void handleDisabled(DlmStatDTO dlmStatDTO) {
        dlmStatDTO.setDayAmount(0L);
        dlmStatDTO.setAvailableAmount(0L);
        dlmStatDTO.setObjectCount(0L);
        dlmStatDTO.setOtherCount(0L);
        dlmStatDTO.setTotalAmount(0L);
    }
}
